package cn.me.alphamq.broker.core.consumequeue;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.constant.BrokerConstant;
import cn.me.alphamq.common.model.broker.meta.MMapFileModel;
import cn.me.alphamq.common.model.broker.CommitLogTopicModel;
import cn.me.alphamq.common.model.broker.ConsumeQueueFileModel;
import cn.me.alphamq.broker.model.consumequeue.ConsumeQueueIndexModel;
import cn.me.alphamq.common.util.FileNameUtil;
import cn.me.alphamq.common.util.AssertUtil;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * ConsumeQueue 的 MMap 文件的模型
 *
 * @author f
 */
public class ConsumeQueueMMapFileModel {

    /**
     * 索引单元的大小为 16 字节
     */
    public static final int INDEX_UNIT_SIZE = 16;

    /**
     * 底层用来读写数据的对象
     */
    private MMapFileModel mMapFileModel;

    /**
     * 本 MMap 文件对应的 topic 名称
     */
    private String topicName;

    /**
     * 本 MMap 文件对应的 队列 id
     */
    private Integer queueId;

    /**
     * 指定 offset 做文件的映射
     *
     * @param topicName 消息的 topic
     * @param queueId   队列 id
     */
    public void loadFileInMMap(String topicName, Integer queueId) throws IOException {
        this.topicName = topicName;
        this.queueId = queueId;

        CommitLogTopicModel commitLogTopicModel = Container.fetchCommitLogTopicModelByTopic(topicName);
        AssertUtil.isNotNull(commitLogTopicModel, "topicName [" + topicName + "] is invalid");

        ConsumeQueueFileModel consumeQueueFileModel = commitLogTopicModel.fetchConsumeQueueFileByQueueId(queueId);
        AssertUtil.isNotNull(consumeQueueFileModel, "consumeQueue [" + queueId + "] is invalid");

        mMapFileModel = new MMapFileModel();
        mMapFileModel.mapFile(consumeQueueFileModel, FileNameUtil.getConsumeQueueFileDir(topicName, queueId),
                BrokerConstant.CONSUME_QUEUE_MAX_SIZE);
    }

    /**
     * 从文件的指定 offset 开始读取多条消息索引内容
     *
     * @param readOffset
     * @param batchMsgNum 读取的消息索引条数
     * @return
     */
    public List<byte[]> readBatchContent(int readOffset, int batchMsgNum) {
        List<byte[]> res = new ArrayList<>();
        for (int i = 0; i < batchMsgNum; i++) {
            res.add(mMapFileModel.read(readOffset + i * INDEX_UNIT_SIZE, INDEX_UNIT_SIZE));
        }
        return res;
    }

    /**
     * 更高性能的写入 api，异步刷盘
     *
     * @param consumeQueueIndexModel
     */
    public void writeContent(ConsumeQueueIndexModel consumeQueueIndexModel) throws Exception {
        writeContent(consumeQueueIndexModel, false);
    }

    /**
     * 写入数据到磁盘当中，可以选择同步还是异步刷盘
     *
     * @param consumeQueueIndexModel
     * @param force
     */
    public void writeContent(ConsumeQueueIndexModel consumeQueueIndexModel, boolean force) throws Exception {
        mMapFileModel.write(consumeQueueIndexModel.toBytes(), force);
    }

    /**
     * 释放 MMap 内存占用
     */
    public void clean() {
        mMapFileModel.clean();
    }

    public Integer getQueueId() {
        return queueId;
    }

    public String getTopicName() {
        return topicName;
    }
}
